package com.talent.carrot.B_workqueues;

import com.rabbitmq.client.*;
import org.yaml.snakeyaml.Yaml;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;

/**
 * 躲了一辈子雨，雨会不会很难过。
 * </p>
 * Created by zhumenglong on 2017/6/12.
 */
public class NewWorker {

    //队列名称
    private final static String TASK_QUEUE_NAME = "task_queue";

    private static InputStream rabbitConfig;

    private static Map configMap;

    static {
        rabbitConfig = Task.class.getClassLoader().getResourceAsStream("RabbitMqConfig.yaml");
    }

    public static void main(String[] args) throws Exception{
        Yaml yaml = new Yaml();
        configMap = (HashMap<String, String>) yaml.load(rabbitConfig);

        //创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost((String) configMap.get("CF.Host"));
        //指定用户 密码
        factory.setUsername((String) configMap.get("CF.Username"));
        factory.setPassword((String) configMap.get("CF.Password"));
        //指定端口
        factory.setPort(AMQP.PROTOCOL.PORT);
        Connection connection = factory.newConnection();
        // 这里之所以声明为final是因为内部类中要访问的变量需要被声明为final
        final Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        /*
         *  prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
         *  global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
         *  备注：据说prefetchSize 和global这两项，rabbitmq没有实现，暂且不研究
         */
        channel.basicQos(1);

        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                System.out.println(" [x] Received '" + message + "'");
                try {
                    doWork(message);
                } finally {
                    System.out.println(" [x] Done");
                    // 返回确认状态。即使未处理消息时工作线程被kill掉，消息也不会丢失。
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
    }

    /**
     * 每个点耗时1s
     * @param task
     * @throws InterruptedException
     */
    private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
